package hy;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Arrays;
import java.util.Properties;

public class MyConsumer {
    public static void main(String[] args) {
        Properties p=new Properties();
        p.put("bootstrap.servers","192.168.184.100:9092");
        p.put("group.id","gr_03");
        p.put("enable.auto.commit","false");

        p.put("key.deserializer", StringDeserializer.class.getName());
        p.put("value.deserializer", StringDeserializer.class.getName());
        p.put("auto.offset.reset","earliest");//从头开始消费

        KafkaConsumer<String,String>consumer=new KafkaConsumer<String, String>(p);
        while(true){
            consumer.subscribe(Arrays.asList("first","second"));
            ConsumerRecords<String,String>cr=consumer.poll(100);
            for(ConsumerRecord<String,String>r:cr){
                System.out.println("主题："+r.topic()+"，所属分区："+r.partition()+",offset:"+r.offset()+",内容："+r.value());
            }
            consumer.commitSync();
        }
        //手动提交
        /*
        * 异步：commitAsync,提交后就结束，不阻塞
        * 同步：commitSync,提交后会等待kafka确认，在收到确认前会阻塞
        * */

    }
}
